package cn.dglydrpy.study.j2ee.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.function.BiConsumer;
/**
* for i in `seq 60`; do echo "hello $i" >> nm.txt; done
* # 创建主题，三个分区
* kafka-topics.sh --zookeeper 192.168.80.131:2181/myKafka --create --topic topic_1 --partitions 3 --replication-factor 1
* # 将消息生产到主题中
* kafka-console-producer.sh --broker-list node1:9092 --topic topic_1 < nm.txt
 *
 */
public class MyKafkaOffsetConsumer {
    public static void main(String[] args) {

        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.80.131:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // group.id很重要
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka_consumer_g1");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
        
// 订阅，执行poll时才会真正进行订阅  自动分区分配
//        consumer.subscribe(Collections.singleton("topic_1"));

        // 如何手动给消费者分配分区？
        // 1、需要知道有哪些主题可以访问和消费

        // listTopics 获取当前消费者可以访问和消费的主题以及它们的分区信息
        final Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();

        stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
            @Override
            public void accept(String topicName, List<PartitionInfo> partitionInfos) {
                System.out.println("主题名称：" + topicName);
                for (PartitionInfo partitionInfo : partitionInfos) {
                    System.out.println(partitionInfo);
                }
            }
        });
        System.out.println("----------------------------");
        
        // partitionsFor 列出某个主题的分区
        List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic_1");
        for (PartitionInfo partitionInfo : partitionInfos) {
        	Node leader = partitionInfo.leader();
        	System.out.println(leader);
        	System.out.println(partitionInfo);
        	// 当前分区在线副本
        	Node[] nodes = partitionInfo.inSyncReplicas();
        	// 当前分区下线副本
        	Node[] nodes1 = partitionInfo.offlineReplicas();
        }
        System.out.println("----------------------------");
        
        // 获取给当前消费者分配的主题分区信息
        final Set<TopicPartition> assignment1 = consumer.assignment();
        for (TopicPartition partition : assignment1) {
            System.out.println(partition);
        }
        System.out.println("----------------------------");

        /* 给当前消费者手动分配一系列主题分区。该操作会覆盖之前的分配。
         * 如果给出的主题分区是空的，则等价于调用unsubscribe方法。
         * 当消费组成员变了，或者集群或主题的元数据改变了，不会触发分区分配的再平衡。
         * 手动分区分配assign(Collection)不能和自动分区分配subscribe(Collection,ConsumerRebalanceListener)一起使用。
         * 如果启用了自动提交偏移量，则在新的分区分配替换旧的分区分配之前，会对旧的分区分配中的消费偏移量进行异步提交。
         */
        consumer.assign(Arrays.asList(
                new TopicPartition("topic_1", 0),
                new TopicPartition("topic_1", 1),
                new TopicPartition("topic_1", 2)
        ));

        
        final Set<TopicPartition> assignment = consumer.assignment();
        for (TopicPartition partition : assignment) {
            System.out.println(partition);
        }
        System.out.println("************************************");
        
        // 查看当前消费者在指定主题的分区上的消费者偏移量
        final long offset = consumer.position(new TopicPartition("topic_1", 0));
        System.out.println("当前主题在0号分区上的位移：" + offset);

        consumer.seekToBeginning(Arrays.asList(
                new TopicPartition("topic_1", 0),
                new TopicPartition("topic_1", 2)
        ));
        long offset0 = consumer.position(new TopicPartition("topic_1", 0));
        long offset1 = consumer.position(new TopicPartition("topic_1", 1));
        long offset2 = consumer.position(new TopicPartition("topic_1", 2));

        System.out.println(offset0);
        System.out.println(offset1);
        System.out.println(offset2);
        System.out.println("=================================================");

        /**
         * 将偏移量移动到每个给定分区的最后一个。
        该方法延迟执行，只有当调用过poll()或position()之后才可以使用。
        如果没有指定分区，则将当前消费者分配的所有分区的消费者偏移量移动到最后。
        如果设置了隔离级别为：isolation.level=read_committed，则会将分区的消费偏移量移动到最后一个稳定的偏移量，即下一个要消费的消息现在还是未提交状态的事务消息。
        
        seekToBeginning 移动到每个给定分区的起始偏移量
         */
        consumer.seekToEnd(Arrays.asList(new TopicPartition("topic_1", 2)));
        
        /**
         * 将给定主题分区的消费偏移量移动到指定的偏移量，即当前消费者下一条要消费的消息偏移量。
        若该方法多次调用，则最后一次的覆盖前面的。
        如果在消费中间随意使用，可能会丢失数据。
         */
        consumer.seek(new TopicPartition("topic_1", 1), 14);

        offset0 = consumer.position(new TopicPartition("topic_1", 0));
        offset1 = consumer.position(new TopicPartition("topic_1", 1));
        offset2 = consumer.position(new TopicPartition("topic_1", 2));

        System.out.println(offset0);
        System.out.println(offset1);
        System.out.println(offset2);

        consumer.close();
    }



}
